package com.ndp.fb.mq.consumer.publish;


import com.alibaba.dubbo.config.annotation.Reference;
import com.ndp.fb.constant.ConfigConst;
import com.ndp.fb.enums.publish.PublishStatus;
import com.ndp.fb.model.business.bo.useraudience.result.TargetingSpecGroupBo;
import com.ndp.fb.model.business.vo.cache.result.PublishCache;
import com.ndp.fb.model.business.vo.cache.result.UserAudienceCache;
import com.ndp.fb.model.fbagent.param.ApiCallBackTarget;
import com.ndp.fb.mq.consumer.AbstractPullConsumer;
import com.ndp.fb.mq.service.publishhandleservice.post.PublishPostUpdateCreativeForAd;
import com.ndp.fb.mq.service.schedulerservice.adpublish.*;
import com.ndp.fb.mq.service.schedulerservice.thirdpictureaudit.AdGroupAuditForThirdPicService;
import com.ndp.fb.mq.service.schedulerservice.thirdpictureaudit.AutoCheckRecourceOneStep;
import com.ndp.fb.mq.service.schedulerservice.thirdpictureaudit.AutoCheckResourceFourStep;
import com.ndp.fb.rdb.api.AudienceManagerRdbService;
import com.ndp.fb.model.rdb.AudienceManager;
import com.ndp.fb.redis.api.RedisClient;
import com.restfb.json.JsonException;
import com.restfb.json.JsonObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * 广告发布消费者，广告发布，用户定位等等
 * Created by infi on 2015/1/21.
 */
@Component
public class AdPublishConsumerImpl extends AbstractPullConsumer {

    private static Logger logger = LoggerFactory.getLogger(AdPublishConsumerImpl.class);

    @Autowired
    private RedisClient redisClient;

    @Autowired
    private PublishAdCreativeService publishAdCreativeService;

    @Autowired
    private PublishPostUpdateCreativeForAd publishPostUpdateCreativeForAd;

    @Autowired
    private PublishAdGroupService publishAdGroupService;

    @Autowired
    private PublishCampaignService publishCampaignGroupService;

    @Autowired
    private PublishAdSetService publishAdSetService;

    @Autowired
    private AdGroupAuditForThirdPicService adGroupAuditForThirdPicService;

    @Autowired
    private AutoCheckResourceFourStep autoCheckResourceFourStep;

    @Autowired
    private AutoCheckRecourceOneStep autoCheckRecourceOneStep;

    @Autowired
    private FetchFbUsersService fbUsersService;

    @Reference
    private AudienceManagerRdbService audienceManagerRdbService;

    @Override
    public String getTopicName() {
        return ConfigConst.MQ_SERVICES_ADPUBLISH_TOPIC;
    }


    @Override
    protected void consume(String tag, String keys, Object message, Map<String, String> equipment) {
        logger.debug("AdPublishConsumerImpl tag >>>> " + tag + " keys >>>" + keys + " message >>> " + message);
        ApiCallBackTarget.AdPublish target = (ApiCallBackTarget.AdPublish) ApiCallBackTarget.getTargetType(tag);
        switch (target) {
            case TargetingSpec:
                publishUserAudience(equipment, message.toString());
                break;
            case TargetingSpecUsers:
                fbUsersService.fetchFbUsersCallback(equipment, message.toString());
                break;
            case AdCreative:
                publishAdCreativeService.callback(equipment, message);
                break;
            case AdCampaignGroup:
                publishCampaignGroupService.callback(equipment, message);
                break;
            case OutlinkAdCreative:
                publishPostUpdateCreativeForAd.createCreativeCallBack(equipment, message);
                break;
            case OutlinkAdGroup:
                publishPostUpdateCreativeForAd.updateAdGroupCallBack(equipment, message);
                break;
            case AuditAd:
                adGroupAuditForThirdPicService.auditAdCallback(equipment, message);
                break;
            case AuditAdFix:
                adGroupAuditForThirdPicService.auditAdFixCallback(equipment, message);
                break;
            case AdSet:
                publishAdSetService.callback(equipment, message);
                break;
            case AdGroup:
                publishAdGroupService.callback(equipment, message);
                break;
            case AutoAdCampaign:
                autoCheckRecourceOneStep.autoPublishAdCampaignCallBack(equipment, message);
                break;
            case AutoAdSet:
                autoCheckRecourceOneStep.autoPublishAdSetCallBack(equipment, message);
                break;
            case AutoAdCreative:
                autoCheckRecourceOneStep.autoPublishAdCreativeCallBack(equipment, message);
                break;
            case AutoAdGroup:
                autoCheckResourceFourStep.autoPublishAdCallBack(equipment, message);
                break;
            case SimpleTargetingSpec:
            	publisSimplehUserAudience(equipment, message.toString());
                break;
            case RefreshAudienceUser:
                refreshAudienceUser(equipment, message.toString());
                break;
            default:
                logger.warn("不识别的消息tag >>>>" + tag + " ; message >>> " + message);
        }
    }

    private void refreshAudienceUser(Map<String, String> equipment, String response) {
        logger.info("PublishUserAudienceJob refreshAudienceUser 调用开始 ");
        Long audienceId = Long.parseLong(equipment.get("id"));
        try {
            JsonObject jsonObject = new JsonObject(response).getJsonObject("data");
            Long people = jsonObject.getLong("users");
            AudienceManager audienceManager = audienceManagerRdbService.findById(audienceId);
            audienceManager.setPotentialUser(people);
            audienceManagerRdbService.update(audienceManager);
            logger.debug(" refreshAudienceUser >>>" + audienceId + " 成功");
        } catch (JsonException e) {
            logger.error(" refreshAudienceUser >>>" + audienceId + " 失败", e);
        } catch (Exception e) {
            logger.error(" userAudienceCacheId >>>"  + audienceId + "失败", e);
        } finally {
            logger.info("PublishUserAudienceJob refreshAudienceUser 调用结束 ");
        }
    }

    private void publisSimplehUserAudience(Map<String, String> equipment, String response) {
    	 logger.debug("PublishUserAudienceJob publishUserAudience 调用开始 ");
         long startTime = System.currentTimeMillis();
         String publishCacheId = equipment.get("publishCacheId");
         String groupId = equipment.get("groupId");
         String userAudienceCacheId = equipment.get("userAudienceCacheId");
         logger.debug("PublishUserAudienceJob publishUserAudience publishCacheId >>>> " + publishCacheId
                 + " ,userAudienceCacheId >>> " + userAudienceCacheId + " ,groupId >>> " + groupId);
         TargetingSpecGroupBo targetingSpecGroupBo = new TargetingSpecGroupBo();
    	 try {
             targetingSpecGroupBo.setTargetingSpecs(equipment.get("targetingSpec"));
             JsonObject jsonObject = new JsonObject(response).getJsonObject("data");
             Long people = jsonObject.getLong("users");
             targetingSpecGroupBo.setPeople(people);
             targetingSpecGroupBo.setStatus(PublishStatus.SUCCEED.getCode());
             logger.debug(" userAudienceCacheId >>> " + userAudienceCacheId + " ,groupId >>> " + groupId + " 成功");
         } catch (JsonException e) {
             logger.error(" userAudienceCacheId >>> " + userAudienceCacheId + " ,groupId >>> " + groupId + " 失败", e);
             targetingSpecGroupBo.setStatus(PublishStatus.FAIL.getCode());
             targetingSpecGroupBo.setException(response);
         } catch (Exception e) {
             logger.error(" userAudienceCacheId >>> " + userAudienceCacheId + " ,groupId >>> " + groupId + " 失败", e);
             targetingSpecGroupBo.setStatus(PublishStatus.FAIL.getCode());
             targetingSpecGroupBo.setException(e.getMessage());
         }finally {
             redisClient.set(userAudienceCacheId, targetingSpecGroupBo);
             redisClient.expire(userAudienceCacheId, 30, TimeUnit.MINUTES.name());
             logger.info("publishUserAudience>>>end>>>>>time>>" + (System.currentTimeMillis() - startTime));
             logger.debug("PublishUserAudienceJob publishUserAudience 回调完成 ");
		}
    }
    /**
     * 处理用户定位
     *
     * @param equipment
     * @param response
     */
    private void publishUserAudience(Map<String, String> equipment, String response) {
        logger.debug("PublishUserAudienceJob publishUserAudience 调用开始 ");
        long startTime = System.currentTimeMillis();
        String publishCacheId = equipment.get("publishCacheId");
        String groupId = equipment.get("groupId");
        String userAudienceCacheId = equipment.get("userAudienceCacheId");
        logger.debug("PublishUserAudienceJob publishUserAudience publishCacheId >>>> " + publishCacheId
                + " ,userAudienceCacheId >>> " + userAudienceCacheId + " ,groupId >>> " + groupId);

        //1、根据publishCacheId获取相应的PublishCache
        PublishCache publishCache = (PublishCache) redisClient.get(publishCacheId);
        logger.info("publishUserAudience>>>redisresule>>>>>time>>" + (System.currentTimeMillis() - startTime));
        startTime = System.currentTimeMillis();
        if (null == publishCache || null == publishCache.getUserAudience()) {
            logger.info("PublishUserAudienceJob publishUserAudience publishCacheId >>>> " + publishCacheId
                    + "不存在!");
            return;
        }

        //2、根据PublishCache中UserAudienceCache的userAudienceCacheId判断是否是同一次用户定位
        UserAudienceCache userAudienceCache = publishCache.getUserAudience();
        if (!userAudienceCacheId.equals(userAudienceCache.getUserAudienceCacheId())) {
            //如果不是放弃本次回调
            logger.info("PublishUserAudienceJob publishUserAudience publishCacheId >>>> " + publishCacheId
                    + "下重新生成用户定位，new userAudienceCacheId >>>> " + userAudienceCache.getUserAudienceCacheId()
                    + "，old userAudienceCacheId >>>> " + userAudienceCacheId);
            return;
        }

        Map<String, TargetingSpecGroupBo> previewTargetingSpecGroups = userAudienceCache.getPreviewTargetingSpecGroups();
        TargetingSpecGroupBo targetingSpecGroupBo = previewTargetingSpecGroups.get(groupId);
        if (null == targetingSpecGroupBo) {
            logger.info("PublishUserAudienceJob publishUserAudience publishCacheId >>>> " + publishCacheId
                    + " ,userAudienceCacheId >>> " + userAudienceCacheId + " ,groupId >>> " + groupId + "不存在!");
            return;
        }

        //3、解析响应
        try {
            JsonObject jsonObject = new JsonObject(response).getJsonObject("data");
            Long people = jsonObject.getLong("users");
            targetingSpecGroupBo.setPeople(people);
            targetingSpecGroupBo.setStatus(PublishStatus.SUCCEED.getCode());
            logger.debug(" userAudienceCacheId >>> " + userAudienceCacheId + " ,groupId >>> " + groupId + " 成功");
        } catch (JsonException e) {
            logger.error(" userAudienceCacheId >>> " + userAudienceCacheId + " ,groupId >>> " + groupId + " 失败", e);
            targetingSpecGroupBo.setStatus(PublishStatus.FAIL.getCode());
            targetingSpecGroupBo.setException(response);
        } catch (Exception e) {
            logger.error(" userAudienceCacheId >>> " + userAudienceCacheId + " ,groupId >>> " + groupId + " 失败", e);
            targetingSpecGroupBo.setStatus(PublishStatus.FAIL.getCode());
            targetingSpecGroupBo.setException(e.getMessage());
        } finally {
            //4、将响应结果存入缓存
            //4.1、创建用户定位组缓存
            String key  = userAudienceCacheId + "_incr";
            String previewUserAudienceGroupKey = buildPreviewUserAudienceGroupKey(publishCacheId, userAudienceCacheId, groupId);
            redisClient.set(previewUserAudienceGroupKey, targetingSpecGroupBo);
            Double incr = redisClient.incr(key);
            redisClient.expire(key, 30, TimeUnit.MINUTES.name());
            logger.info("已经完成key>>>>" + key + "  ,incr>>>>" + incr);

            //4.2、判断本次用户定位是否全部完成
            updateCompleteNew(publishCacheId, publishCache, incr);
//            updateComplete(publishCacheId, publishCache);
            logger.info("publishUserAudience>>>end>>>>>time>>" + (System.currentTimeMillis() - startTime));
            logger.debug("PublishUserAudienceJob publishUserAudience 回调完成 ");
        }
    }

    private void updateCompleteNew(String publishCacheId, PublishCache publishCache, Double incr) {
        UserAudienceCache userAudienceCache = publishCache.getUserAudience();
        String userAudienceCacheId = userAudienceCache.getUserAudienceCacheId();
        Map<String, TargetingSpecGroupBo> previewTargetingSpecGroups = userAudienceCache.getPreviewTargetingSpecGroups();
        if(incr!= null && incr.intValue() == previewTargetingSpecGroups.size()){
            logger.debug("publishCacheId >>>> " + publishCacheId + " ,cacheId >>> " + userAudienceCacheId + " >>> 预览用户定位完成");

            List<TargetingSpecGroupBo> publishingGroups = new ArrayList<TargetingSpecGroupBo>();
            Map<String, TargetingSpecGroupBo> finalPreviewTargetingSpecGroups = new HashMap<String, TargetingSpecGroupBo>();
            Set<String> previewUserAudienceGroupKeys = new HashSet<String>();
            for (Map.Entry<String, TargetingSpecGroupBo> entry : previewTargetingSpecGroups.entrySet()) {
                String previewUserAudienceGroupKey = buildPreviewUserAudienceGroupKey(publishCacheId, userAudienceCacheId, entry.getKey());
                TargetingSpecGroupBo targetingSpecGroupBo = (TargetingSpecGroupBo) redisClient.get(previewUserAudienceGroupKey);
                if (targetingSpecGroupBo != null) {
                    logger.debug("updateCompleteKey> " + previewUserAudienceGroupKey + " ->状态->" + targetingSpecGroupBo.getStatus());
                    previewUserAudienceGroupKeys.add(previewUserAudienceGroupKey);
                    finalPreviewTargetingSpecGroups.put(entry.getKey(), targetingSpecGroupBo);
                    publishingGroups.add(targetingSpecGroupBo);
                }
            }

            userAudienceCache.setIsComplete(true);
            userAudienceCache.setPreviewTargetingSpecGroups(finalPreviewTargetingSpecGroups);
            userAudienceCache.setPublishingTargetUserAudienceResults(publishingGroups);
            userAudienceCache.setPublishingTargetingSpecGroups(publishingGroups);
            redisClient.set(publishCacheId, publishCache);
            redisClient.delete(previewUserAudienceGroupKeys);
            String key  = userAudienceCacheId + "_incr";
            redisClient.delete(key);
        }
    }

    private String buildPreviewUserAudienceGroupKey(String publishCacheId, String userAudienceCacheId, String groupId) {
        return "previewUserAudience" + "_" + publishCacheId + "_" + userAudienceCacheId + "_" + groupId;
    }
}
